package com.lengxf.flink.source;

import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.common.typeinfo.Types;
import org.apache.flink.streaming.api.datastream.*;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.ProcessFunction;
import org.apache.flink.streaming.api.functions.co.CoMapFunction;
import org.apache.flink.util.Collector;
import org.apache.flink.util.OutputTag;

/**
 * 分区测试demo
 *
 * @author Lengxf
 */
public class SplitByDemo {

    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(2);
        DataStreamSource<String> lineDS = env.readTextFile("flink/src/main/resources/number.txt");

        //打印分流区分
        //lineDS.map((MapFunction<String, Integer>) Integer::parseInt)
        //        .returns(Integer.class)
        //        .filter(v -> v % 2 == 0)
        //        .print("偶数流");
        //
        //lineDS.map((MapFunction<String, Integer>) Integer::parseInt)
        //        .returns(Integer.class)
        //        .filter(v -> v % 2 == 1)
        //        .print("奇数流");

        System.out.println("--------------------------");
        OutputTag<Integer> outputTag1 = new OutputTag<>("偶数流", Types.INT);
        OutputTag<Integer> outputTag2 = new OutputTag<>("奇数流", Types.INT);
        SingleOutputStreamOperator<Integer> process = lineDS.map((MapFunction<String, Integer>) Integer::parseInt)
                .process(new ProcessFunction<>() {
                    @Override
                    public void processElement(Integer value, ProcessFunction<Integer, Integer>.Context ctx, Collector<Integer> out) {
                        if (value % 2 == 0) {
                            ctx.output(outputTag1, value);
                        } else {
                            ctx.output(outputTag2, value);
                        }
                    }
                });

        //侧边流输出
        process.getSideOutput(outputTag1).print("偶数流");
        process.getSideOutput(outputTag2).print("奇数流");

        //数据流合并
        SideOutputDataStream<Integer> sideOutput1 = process.getSideOutput(outputTag1);
        DataStream<Integer> sideOutput2 = process.getSideOutput(outputTag2);
        //可以进行多次合并
        sideOutput1.union(sideOutput2).union(sideOutput2).print();
        ConnectedStreams<Integer, Integer> connect = sideOutput1.connect(sideOutput2);
        connect.map(new CoMapFunction<Integer, Integer, String>() {
            @Override
            public String map1(Integer value) throws Exception {
                return "来源于 ： map1 =>" + value;
            }

            @Override
            public String map2(Integer value) throws Exception {
                return "来源于 ： map2 => " + value;
            }
        }).print();

        env.execute();
    }


}
